package componentInit

import (
	"fmt"
	"db2s/global"
	mq "db2s/topic-mq"
)

func (c *MQConfig) CreateMq(listeners map[CheckMode][]*MQTopicListeners) *mq.Mq {
	config := &mq.MqConfig{
		ToPicConfigs:         c.mergerTopicConfig(listeners),
		ConsumerConfig:       c.toConsumerConfig(listeners),
		DefaultTopicPoolSize: 400,
	}
	start, err := config.NewStart()
	if err != nil {
		fmt.Println("Mq Start() err", err)
		return nil
	}
	global.GetIoc().RegisterContext("consumer", start.Consumer)
	global.GetIoc().RegisterContext("producer", start.Producer)
	global.GetIoc().RegisterContext("monitorListener", start.MonitorListener)
	global.GetIoc().RegisterContext("checkType", listeners)
	return start
}
func NewDefaultMq(v ...any) *MQConfig {
	var customizePoolSize int
	if len(v) == 0 {
		customizePoolSize = DefaultConsumerPoolSize
	} else if len(v) == 1 {
		customizePoolSize = v[0].(int)
	} else {
		WLog.Error("Failed to set task concurrency")
		return nil
	}
	return &MQConfig{
		PoolSize: customizePoolSize,
	}
}

// InitAutoCloseMqComponentTable 初始化 mq 自动关闭组件
func InitAutoCloseMqComponentTable() map[CheckMode]*AutoCloseMqComponent {
	m := make(map[CheckMode]*AutoCloseMqComponent)
	checkType = global.GetIoc().GetBean("checkType").(map[CheckMode][]*MQTopicListeners)
	for key, l := range checkType {
		m[key] = initAutoCloseMqComponent(key, l)
	}
	return m
}

func initAutoCloseMqComponent(checkMode CheckMode, check []*MQTopicListeners) *AutoCloseMqComponent {
	m := make(map[Topic]*int64)
	for _, l := range check {
		var i int64 = 0
		m[l.TopicName] = &i
	}
	return &AutoCloseMqComponent{
		TopicStatistics:  m,
		UnsubscribeTopic: make([]Topic, 0),
		CheckMode:        checkMode,
		EndMark:          false,
	}
}

func NewMQTopicListeners(topicName Topic, messageQueueLength int, PoolSize int, Listeners func(message mq.Message) mq.ConsumeConcurrentlyStatus, messageCapLength int, order int) *MQTopicListeners {
	return &MQTopicListeners{
		TopicName:          topicName,
		Listeners:          Listeners,
		MessageQueueLength: messageQueueLength,
		MessageCapLength:   messageCapLength,
		Order:              order,
		PoolSize:           PoolSize,
	}
}

func (c *MQConfig) toConsumerConfig(mt map[CheckMode][]*MQTopicListeners) *mq.ConsumerConfig {
	messageListeners := make(map[string] /*topicName*/ func(message mq.Message) mq.ConsumeConcurrentlyStatus)
	for _, ls := range mt {
		for _, l := range ls {
			messageListeners[string(l.TopicName)] = l.Listeners
		}
	}
	return &mq.ConsumerConfig{
		PoolSize:         c.PoolSize,
		MessageListeners: messageListeners,
	}
}

func (c *MQConfig) toToPicConfig(mt []*MQTopicListeners) []*mq.ToPicConfig {
	topicConfig := make([]*mq.ToPicConfig, 0)
	if len(c.TopicConfig) != 0 {
		for _, config := range c.TopicConfig {
			topicConfig = append(topicConfig, &mq.ToPicConfig{
				TopicName:          string(config.Topic),
				MessageQueueLength: config.MessageQueueLength,
				MessageCapLength:   config.MessageQueueLength,
				Selector:           config.Selector,
			})
		}
		return topicConfig
	}

	for _, listeners := range mt {
		topicConfig = append(topicConfig, &mq.ToPicConfig{
			TopicName: string(listeners.TopicName),
		})
	}
	return topicConfig
}

func (c *MQConfig) mergerTopicConfig(mt map[CheckMode][]*MQTopicListeners) []*mq.ToPicConfig {
	topicConfig := make([]*mq.ToPicConfig, 0)
	for _, ls := range mt {
		for _, l := range ls {
			config := c.TopicConfig[l.TopicName]
			if config == nil {
				topicConfig = append(topicConfig, &mq.ToPicConfig{
					TopicName:          string(l.TopicName),
					MessageQueueLength: l.MessageQueueLength,
					MessageCapLength:   l.MessageCapLength,
					PoolSize:           l.PoolSize,
				})
			} else {
				topicConfig = append(topicConfig, &mq.ToPicConfig{
					TopicName:          string(config.Topic),
					MessageQueueLength: config.GetMessageQueueLength(l.MessageQueueLength),
					MessageCapLength:   config.GetMessageCapLength(l.MessageCapLength),
					Selector:           config.Selector,
				})
			}
		}
	}
	return topicConfig
}

func (c *TopicConfig) GetMessageQueueLength(def int) int {
	if c.MessageQueueLength <= 0 {
		return def
	}
	return c.MessageQueueLength
}

func (c *TopicConfig) GetMessageCapLength(def int) int {
	if c.MessageCapLength <= 0 {
		return def
	}
	return c.MessageCapLength
}
